package spring.cloud.tasks.tasks_manager.schedule;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.api.transaction.CuratorTransaction;
import org.apache.curator.framework.api.transaction.CuratorTransactionFinal;
import org.apache.curator.framework.api.transaction.CuratorTransactionResult;
import spring.cloud.tasks.common.*;
import spring.cloud.tasks.common.curator.Curator;
import spring.cloud.tasks.common.nodepaths.RenewalNodePaths;
import spring.cloud.tasks.common.nodepaths.TaskInstanceNodePaths;
import spring.cloud.tasks.common.status.TaskInstanceInfoStatus;
import spring.cloud.tasks.common.status.TaskItemStatus;
import spring.cloud.tasks.common.status.TaskStageStatus;
import spring.cloud.tasks.tasks_manager.dto.TaskDto;
import spring.cloud.tasks.tasks_manager.taskInstanceManager.BatchTaskTaskInstanceManager;
import spring.cloud.tasks.tasks_manager.taskInstanceManager.StateFullSimpleTaskTaskInstanceManager;
import spring.cloud.tasks.tasks_manager.taskInstanceManager.StateLessSimpleTaskTaskInstanceManager;
import spring.cloud.tasks.tasks_manager.taskInstanceManager.TaskInstanceManager;

import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;

@Slf4j
public class TaskInstanceRegisterManagerUtil {
    public static TaskInstanceInfo buildTaskInstanceInfo(TaskDto taskDto, FireDateTimeAndPreAssignDateTime fireDateTimeAndPreAssignDateTime) {
        TaskInstanceInfo taskInstanceInfo = new TaskInstanceInfo();
        taskInstanceInfo.setRunType(TaskInstanceInfo.RUN_TYPE_TIME_SCHEDULE);
        taskInstanceInfo.setTaskType(fireDateTimeAndPreAssignDateTime.getTaskType());
        taskInstanceInfo.setTaskId(fireDateTimeAndPreAssignDateTime.getTaskId());
        taskInstanceInfo.setTaskLevel(taskDto.getTaskLevel());
        taskInstanceInfo.setVersion(fireDateTimeAndPreAssignDateTime.getFireDateTimeVersion());
        taskInstanceInfo.setNextPreAssignDate(fireDateTimeAndPreAssignDateTime.getNextPreAssignDate());
        taskInstanceInfo.setNextFireDate(fireDateTimeAndPreAssignDateTime.getNextFireDate());
        taskInstanceInfo.setTimeoutMinutes(taskDto.getTimeoutMinutes());
        taskInstanceInfo.setLockMinutes(taskDto.getLockMinutes());
        taskInstanceInfo.setClearMinutes(taskDto.getClearMinutes());
        //
        taskInstanceInfo.setTaskProperties(taskDto.getTaskProperties());
        taskInstanceInfo.setTaskDataRangeRule(taskDto.getTaskDataRangeRule());
        taskInstanceInfo.setTaskItemInfo(taskDto.getTaskItemInfo());
        //
        taskInstanceInfo.setTaskInstanceId(getTaskInstanceId());
        taskInstanceInfo.setStatus(TaskInstanceInfoStatus.create_instance_object);
        return taskInstanceInfo;
    }

    public static TaskInstanceInfo buildUserSubmitRequestTaskInstanceInfo(TaskDto taskDto, UserSubmitRequestTaskInstance userSubmitRequestTaskInstance) {
        TaskInstanceInfo taskInstanceInfo = new TaskInstanceInfo();
        taskInstanceInfo.setRunType(TaskInstanceInfo.RUN_TYPE_RIGHT_NOW_RUN);
        taskInstanceInfo.setTaskType(taskDto.getTaskType());
        taskInstanceInfo.setTaskId(taskDto.getTaskId());
        taskInstanceInfo.setTaskLevel(taskDto.getTaskLevel());
        taskInstanceInfo.setVersion(taskDto.getFireDateTimeVersion());
        taskInstanceInfo.setNextPreAssignDate(new Date());
        taskInstanceInfo.setNextFireDate(new Date());
        taskInstanceInfo.setTimeoutMinutes(taskDto.getTimeoutMinutes());
        taskInstanceInfo.setLockMinutes(taskDto.getLockMinutes());
        taskInstanceInfo.setClearMinutes(taskDto.getClearMinutes());
        //
        taskInstanceInfo.setTaskProperties(taskDto.getTaskProperties());
        taskInstanceInfo.setTaskDataRangeRule(taskDto.getTaskDataRangeRule());
        taskInstanceInfo.setTaskItemInfo(taskDto.getTaskItemInfo());
        //
        taskInstanceInfo.setTaskInstanceId(getTaskInstanceId());
        taskInstanceInfo.setStatus(TaskInstanceInfoStatus.create_instance_object);
        return taskInstanceInfo;
    }

    //
    public static TaskInstanceInfo buildRightNowRunTaskInstanceInfo(TaskDto taskDto, RightNowRunTaskInstance rightNowRunTaskInstance) {
        TaskInstanceInfo taskInstanceInfo = new TaskInstanceInfo();
        taskInstanceInfo.setRunType(TaskInstanceInfo.RUN_TYPE_RIGHT_NOW_RUN);
        taskInstanceInfo.setTaskType(taskDto.getTaskType());
        taskInstanceInfo.setTaskId(taskDto.getTaskId());
        taskInstanceInfo.setTaskLevel(taskDto.getTaskLevel());
        taskInstanceInfo.setVersion(taskDto.getFireDateTimeVersion());
        taskInstanceInfo.setNextPreAssignDate(new Date());
        taskInstanceInfo.setNextFireDate(new Date());
        taskInstanceInfo.setTimeoutMinutes(taskDto.getTimeoutMinutes());
        taskInstanceInfo.setLockMinutes(taskDto.getLockMinutes());
        taskInstanceInfo.setClearMinutes(taskDto.getClearMinutes());
        //
        taskInstanceInfo.setTaskProperties(taskDto.getTaskProperties());
        taskInstanceInfo.setTaskDataRangeRule(taskDto.getTaskDataRangeRule());
        taskInstanceInfo.setTaskItemInfo(taskDto.getTaskItemInfo());
        //
        taskInstanceInfo.setTaskInstanceId(getTaskInstanceId());
        taskInstanceInfo.setStatus(TaskInstanceInfoStatus.create_instance_object);
        return taskInstanceInfo;
    }

    public static TaskInstanceInfo buildReRunTaskInstanceInfo(TaskDto taskDto, ReRunTaskInstance reRunTaskInstance) {
        TaskInstanceInfo taskInstanceInfo = new TaskInstanceInfo();
        taskInstanceInfo.setRunType(TaskInstanceInfo.RUN_TYPE_RIGHT_NOW_RUN);
        taskInstanceInfo.setTaskType(taskDto.getTaskType());
        taskInstanceInfo.setTaskId(taskDto.getTaskId());
        taskInstanceInfo.setTaskLevel(taskDto.getTaskLevel());
        taskInstanceInfo.setVersion(taskDto.getFireDateTimeVersion());
        taskInstanceInfo.setNextPreAssignDate(new Date());
        taskInstanceInfo.setNextFireDate(new Date());
        taskInstanceInfo.setTimeoutMinutes(taskDto.getTimeoutMinutes());
        taskInstanceInfo.setLockMinutes(taskDto.getLockMinutes());
        taskInstanceInfo.setClearMinutes(taskDto.getClearMinutes());
        //
        taskInstanceInfo.setTaskProperties(taskDto.getTaskProperties());
        taskInstanceInfo.setTaskDataRangeRule(taskDto.getTaskDataRangeRule());
        taskInstanceInfo.setTaskItemInfo(taskDto.getTaskItemInfo());
        //
        taskInstanceInfo.setOldTaskInstanceId(reRunTaskInstance.getTaskInstanceId());
        taskInstanceInfo.setCurrentStage(reRunTaskInstance.getCurrentStage());
        taskInstanceInfo.setCurrentStageStatus(TaskStageStatus.status_watting_init);
        //
        taskInstanceInfo.setTaskInstanceId(getTaskInstanceId());
        taskInstanceInfo.setStatus(TaskInstanceInfoStatus.create_instance_object);
        return taskInstanceInfo;
    }

    /**
     * 这样生成的实例id内存占用量就小了
     *
     * @return
     */
    public static synchronized String getTaskInstanceId() {
        try {
            TimeUnit.MILLISECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        final String dateFormate = "yyyyMMdd:HHmmssSSS";
        return new SimpleDateFormat(dateFormate).format(new Date());
    }

    public static void taskInstanceInfoRegisterAtZk(Curator curator, TaskInstanceInfo taskInstanceInfo) throws Exception {
        //单个事务保证
        String taskInstanceBaseInfoNodePath = TaskInstanceNodePaths.getTaskInstanceSubNodePath(taskInstanceInfo.getTaskId(), taskInstanceInfo.getTaskInstanceId(), TaskInstanceNodePaths.TaskInstanceSubNode.baseInfo);
        String taskInstanceCreateTaskItemsNodePath = TaskInstanceNodePaths.getTaskInstanceSubNodePath(taskInstanceInfo.getTaskId(), taskInstanceInfo.getTaskInstanceId(), TaskInstanceNodePaths.TaskInstanceSubNode.createTaskItems);
        String taskInstanceScheduleNodePath = TaskInstanceNodePaths.getTaskInstanceSubNodePath(taskInstanceInfo.getTaskId(), taskInstanceInfo.getTaskInstanceId(), TaskInstanceNodePaths.TaskInstanceSubNode.schedule);
        String taskInstanceStatusNodePath = TaskInstanceNodePaths.getTaskInstanceSubNodePath(taskInstanceInfo.getTaskId(), taskInstanceInfo.getTaskInstanceId(), TaskInstanceNodePaths.TaskInstanceSubNode.status);
        if (curator.checkExists(taskInstanceBaseInfoNodePath)) {
            return;
        }
        if (curator.checkExists(taskInstanceStatusNodePath)) {
            return;
        }
        TaskInstanceBaseInfo taskInstanceBaseInfo = new TaskInstanceBaseInfo();
        taskInstanceBaseInfo.copy(taskInstanceInfo);
        CuratorTransaction curatorTransaction = curator.getCuratorFramework().inTransaction();
        CuratorTransactionFinal curatorTransactionFinal = (CuratorTransactionFinal) curatorTransaction;
        curatorTransactionFinal.setData().forPath(taskInstanceBaseInfoNodePath, (JSONObject.toJSONString(taskInstanceBaseInfo)).getBytes(StandardCharsets.UTF_8)).and();
        curatorTransactionFinal.setData().forPath(taskInstanceCreateTaskItemsNodePath, YN.N.getBytes(StandardCharsets.UTF_8)).and();
        curatorTransactionFinal.setData().forPath(taskInstanceScheduleNodePath, YN.N.getBytes(StandardCharsets.UTF_8)).and();
        curatorTransactionFinal.setData().forPath(taskInstanceStatusNodePath, TaskInstanceInfoStatus.$register_task_instance_info_success$.getBytes(StandardCharsets.UTF_8)).and();
        try {
            Collection<CuratorTransactionResult> commit = curatorTransactionFinal.commit();
            for (CuratorTransactionResult curatorTransactionResult : commit) {
                System.out.println(curatorTransactionResult);
            }
            taskInstanceInfo.setStatus(TaskInstanceInfoStatus.$register_task_instance_info_success$);
            taskInstanceInfo.setRegistered(true);
            return;
        } catch (Throwable e) {
            //抛出异常则为创建失败
            taskInstanceInfo.setStatus(TaskInstanceInfoStatus.$register_task_instance_info_failure$);
            return;
        }
    }

    public static Boolean taskInstanceInfoRegisterAtZkAndCreateTaskItemsAtZk(ScheduleManager scheduleManager, Curator curator, TaskInstanceInfo taskInstanceInfo, TaskDto taskDto) throws Exception {
        TaskInstanceRegisterManagerUtil.taskInstanceInfoRegisterAtZk(curator, taskInstanceInfo);
        while (TaskInstanceInfoStatus.$create_instance_task_item_info_failure$.equals(taskInstanceInfo.getStatus())) {
            taskInstanceInfo.setTaskInstanceId(TaskInstanceRegisterManagerUtil.getTaskInstanceId());
            TaskInstanceRegisterManagerUtil.taskInstanceInfoRegisterAtZk(curator, taskInstanceInfo);
            if (TaskInstanceInfoStatus.$create_instance_task_item_info_success$.equals(taskInstanceInfo.getStatus())) {
                break;
            }
        }
        try {
            /**
             * 支持重复创建
             */
            TaskInstanceRegisterManagerUtil.createTaskItemsAtZk(taskDto, curator, taskInstanceInfo);
            String taskInstanceStatusNodePath = TaskInstanceNodePaths.getTaskInstanceSubNodePath(taskInstanceInfo.getTaskId(), taskInstanceInfo.getTaskInstanceId(), TaskInstanceNodePaths.TaskInstanceSubNode.status);
            curator.merge(taskInstanceStatusNodePath, TaskInstanceInfoStatus.$create_instance_task_item_info_success$);
            taskInstanceInfo.setStatus(TaskInstanceInfoStatus.$create_instance_task_item_info_success$);
            taskInstanceInfo.setRegisterErrorInfo("");
            TaskInstanceRegisterManagerUtil.addWaittingAssignedTask(scheduleManager, taskInstanceInfo);
        } catch (Exception e) {
            //失败了一直重试  直到时间过了任务触发执行时间超时。如果重试都不成功，那么只需要DB记录失败，人工手动进行重跑。
            taskInstanceInfo.setTryCreateInstanceInfoTime(taskInstanceInfo.getTryCreateInstanceInfoTime() + 1);
            taskInstanceInfo.setRegisterErrorInfo(e.getMessage());
            taskInstanceInfo.setStatus(TaskInstanceInfoStatus.$create_instance_task_item_info_failure$);
            return false;
        }
        return true;
    }

    /**
     * ZK上面创建任务分片信息,支持反复执行。
     */
    public static void createTaskItemsAtZk(TaskDto taskDto, Curator curator, TaskInstanceInfo taskInstanceInfo) throws Exception {
        String taskType = taskDto.getTaskType();
        String taskItemInfo = taskDto.getTaskItemInfo();
        //
        String taskInstanceCreateTaskItemsNodePath = TaskInstanceNodePaths.getTaskInstanceSubNodePath(taskInstanceInfo.getTaskId(), taskInstanceInfo.getTaskInstanceId(), TaskInstanceNodePaths.TaskInstanceSubNode.createTaskItems);
        String createTaskItems = curator.getData(taskInstanceCreateTaskItemsNodePath);
        if (YN.Y.equals(createTaskItems)) {
            return;//这样就不用重新创建了
        }
        String rootPath = RenewalNodePaths.getRoot() + Symbols.SLASH + taskInstanceInfo.getTaskId() + Symbols.SLASH + taskInstanceInfo.getTaskInstanceId();
        if (TaskType.SIMPLE_SLICING.getTaskType().equals(taskType)) {
            SimpleSlicingTaskTaskItemCountAndLoadAndParameterInfo simpleSlicingTaskTaskItemCountAndLoadAndParameterInfo = JSON.parseObject(taskItemInfo, SimpleSlicingTaskTaskItemCountAndLoadAndParameterInfo.class);
            int taskItemCount = simpleSlicingTaskTaskItemCountAndLoadAndParameterInfo.taskItemCount;
            int taskItemLoad = simpleSlicingTaskTaskItemCountAndLoadAndParameterInfo.taskItemLoad;
            String taskItemIndexAndTaskItemParametersString = simpleSlicingTaskTaskItemCountAndLoadAndParameterInfo.getTaskItemIndexAndTaskItemParametersString();
            String taskItemPropertiesString = simpleSlicingTaskTaskItemCountAndLoadAndParameterInfo.getTaskItemPropertiesString();
            Map<String, String> taskItemIndexToTaskItemParameterMap = new HashMap<>();
            for (Object o : JSON.parseArray(taskItemIndexAndTaskItemParametersString)) {
                JSONObject jsonObject = (JSONObject) o;
                int taskItemIndex = jsonObject.getIntValue("taskItemIndex");
                String taskItemParameter = jsonObject.getString("taskItemParameter");
                taskItemIndexToTaskItemParameterMap.put(String.valueOf(taskItemIndex), taskItemParameter);
            }
            //简单分片
            for (int i = 1; i <= taskItemCount; i++) {
                String path = rootPath + Symbols.SLASH + TaskType.SimpleSlicingTask.stage_one + Symbols.SLASH + (YN.Y.equals(taskInstanceInfo.getStateFullTask()) ? TaskType.SimpleSlicingTask.stage_type_state_full : TaskType.SimpleSlicingTask.stage_type_state_less) + Symbols.SLASH + i;
                //
                String baseInfoPath = path + Symbols.SLASH + "base-info";
                String statusPath = path + Symbols.SLASH + "status";
                //String markPath = path + Symbols.SLASH + "mark";
                //
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("taskItemParameter", taskItemIndexToTaskItemParameterMap.get(String.valueOf(i)));
                jsonObject.put("taskItemPropertiesString", taskItemPropertiesString);
                curator.createIfAbsent(baseInfoPath, JSON.toJSONString(jsonObject));
                //
                curator.createIfAbsent(statusPath, TaskItemStatus.status_init);//进行状态变更
            }
            curator.merge(taskInstanceCreateTaskItemsNodePath, YN.Y);
            return;
        } else if (TaskType.TIMING_BASED_BATCH_PROCESSING.getTaskType().equals(taskType) || TaskType.USER_TIMING_BASED_BATCH_PROCESSING.getTaskType().equals(taskType)) {
            //批次
            List<TaskStageInfo> taskStageAndTaskItemTypeAndCountAndLoadInfoList = JSON.parseArray(taskItemInfo, TaskStageInfo.class);
            for (TaskStageInfo taskStageAndTaskItemTypeAndCountAndLoadInfo : taskStageAndTaskItemTypeAndCountAndLoadInfoList) {
                String stage = taskStageAndTaskItemTypeAndCountAndLoadInfo.stage;
                List<TaskStageInfo.TaskItemInfo> taskItemTypeAndCountAndLoadList = taskStageAndTaskItemTypeAndCountAndLoadInfo.taskItemInfoList;
                for (TaskStageInfo.TaskItemInfo taskItemTypeAndCountAndLoad : taskItemTypeAndCountAndLoadList) {
                    String type = taskItemTypeAndCountAndLoad.taskItemType;
                    int taskItemCount = taskItemTypeAndCountAndLoad.taskItemCount;
                    int taskItemLoad = taskItemTypeAndCountAndLoad.taskItemLoad;
                    for (int i = 1; i <= taskItemCount; i++) {
                        String path = rootPath + Symbols.SLASH + stage + Symbols.SLASH + type + Symbols.SLASH + i;
                        String baseInfoPath = path + Symbols.SLASH + "base-info";
                        String statusPath = path + Symbols.SLASH + "status";
                        //String markPath = path + Symbols.SLASH + "mark";
                        //
                        JSONObject jsonObject = new JSONObject();
                        curator.createIfAbsent(baseInfoPath, JSON.toJSONString(jsonObject));
                        //
                        curator.createIfAbsent(statusPath, TaskItemStatus.status_init);//进行状态变更
                    }
                }
            }
            curator.merge(taskInstanceCreateTaskItemsNodePath, YN.Y);
            return;
        } else {
            throw new IllegalStateException();
        }
    }


    /**
     * 采用fast select cache模式，不用加锁，都是一个对象(单例变体)
     *
     * @param scheduleManager
     * @param taskTaskInstanceManagerContext
     * @param taskType
     * @param stateFullTask
     * @return
     */
    public static TaskInstanceManager getTaskInstanceManager(ScheduleManager scheduleManager, TaskInstanceRegisterManager.TaskTaskInstanceManagerContext taskTaskInstanceManagerContext, String taskType, String stateFullTask) {
        if (TaskType.SIMPLE_SLICING.getTaskType().equals(taskType)) {
            if (YN.Y.equals(stateFullTask)) {
                if (taskTaskInstanceManagerContext.stateFullSimpleTaskTaskInstanceManager != null) {
                    return taskTaskInstanceManagerContext.stateFullSimpleTaskTaskInstanceManager;
                }
                StateFullSimpleTaskTaskInstanceManager stateFullSimpleTaskTaskInstanceManager = scheduleManager.stateFullSimpleTaskTaskInstanceManager;
                taskTaskInstanceManagerContext.stateFullSimpleTaskTaskInstanceManager = stateFullSimpleTaskTaskInstanceManager;
                return stateFullSimpleTaskTaskInstanceManager;
            } else {
                if (taskTaskInstanceManagerContext.stateLessSimpleTaskTaskInstanceManager != null) {
                    return taskTaskInstanceManagerContext.stateLessSimpleTaskTaskInstanceManager;
                }
                StateLessSimpleTaskTaskInstanceManager stateLessSimpleTaskTaskInstanceManager = scheduleManager.stateLessSimpleTaskTaskInstanceManager;
                taskTaskInstanceManagerContext.stateLessSimpleTaskTaskInstanceManager = stateLessSimpleTaskTaskInstanceManager;
                return stateLessSimpleTaskTaskInstanceManager;
            }
        } else if (TaskType.TIMING_BASED_BATCH_PROCESSING.getTaskType().equals(taskType) || TaskType.USER_TIMING_BASED_BATCH_PROCESSING.getTaskType().equals(taskType)) {
            if (taskTaskInstanceManagerContext.batchTaskTaskInstanceManager != null) {
                return taskTaskInstanceManagerContext.batchTaskTaskInstanceManager;
            }
            BatchTaskTaskInstanceManager batchTaskTaskInstanceManager = scheduleManager.batchTaskTaskInstanceManager;
            taskTaskInstanceManagerContext.batchTaskTaskInstanceManager = batchTaskTaskInstanceManager;
            return batchTaskTaskInstanceManager;
        } else {
            throw new IllegalStateException();
        }
    }

    public static void addWaittingAssignedTask(ScheduleManager scheduleManager, TaskInstanceInfo taskInstanceInfo) {
        if (!TaskInstanceInfoStatus.$create_instance_task_item_info_success$.equals(taskInstanceInfo.getStatus())) {
            return;
        }
        if (taskInstanceInfo.getTaskType().equals(TaskType.TIMING_BASED_BATCH_PROCESSING.getTaskType()) || taskInstanceInfo.getTaskType().equals(TaskType.USER_TIMING_BASED_BATCH_PROCESSING.getTaskType())) {
            ConcurrentHashMap<String, CopyOnWriteArrayList<TaskInstanceInfo>> waittingTaskList = scheduleManager.batchTaskTaskInstanceManager.waitingAssignTasks;
            waittingTaskList.put(taskInstanceInfo.getTaskId(), new CopyOnWriteArrayList<>());
            waittingTaskList.get(taskInstanceInfo.getTaskId()).add(taskInstanceInfo);
        } else if (taskInstanceInfo.getTaskType().equals(TaskType.SIMPLE_SLICING.getTaskType())) {
            if (YN.N.equals(taskInstanceInfo.getStateFullTask())) {
                scheduleManager.stateLessSimpleTaskTaskInstanceManager.waitingAssignTasks.putIfAbsent(taskInstanceInfo.getTaskId(), new CopyOnWriteArrayList<>());
                scheduleManager.stateLessSimpleTaskTaskInstanceManager.waitingAssignTasks.get(taskInstanceInfo.getTaskId()).add(taskInstanceInfo);
            } else if (YN.Y.equals(taskInstanceInfo.getStateFullTask())) {
                //如果是有状态的任务那么也是这样做，具体在进行预调度的时候，会进行任务实例信息的合并处理
                scheduleManager.stateFullSimpleTaskTaskInstanceManager.waitingAssignTasks.putIfAbsent(taskInstanceInfo.getTaskId(), new CopyOnWriteArrayList<>());
                scheduleManager.stateFullSimpleTaskTaskInstanceManager.waitingAssignTasks.get(taskInstanceInfo.getTaskId()).add(taskInstanceInfo);
            } else {
                throw new IllegalStateException();
            }
        } else {
            throw new IllegalStateException();
        }
    }


}
